package cn.itcast.flink.window.time;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
/**
* 窗口统计案例演示：滑动时间窗口（Sliding Time Window），实时交通卡口车流量统计
*/
public class StreamSlidingTimeWindow_3 { @Data
@AllArgsConstructor
@NoArgsConstructor
public static class CartInfo {
private String sensorId;
private Integer count;
}
public static void main(String[] args) throws Exception {
// 1. 执行环境-env：流计算执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1) ;
// 2. 数据源-source：Socket接收数据
DataStreamSource<String> inputStream = env.socketTextStream("node1.itcast.cn", 9999);
/*
数据源：信号灯编号和通过该信号灯的车的数量
a,3
a,2
a,7
d,9
b,6
a,5
b,3
e,7
e,4
*/
// 3. 转换处理-transformation：调用DataSet函数，处理数据
    SingleOutputStreamOperator<CartInfo> carInfoStream = inputStream
// a. 过滤数据
            .filter(line -> null != line && line.trim().length() > 0)
            //解析数据 存储至JavaBean对象
            .map(new MapFunction<String, CartInfo>() {
                @Override
                public CartInfo map(String line) throws Exception {
                    String[] arr = line.trim().split(",");
                    return new CartInfo(arr[0], Integer.parseInt(arr[1]));
                }
            });
    //TODO 按照信号灯编号分组
    SingleOutputStreamOperator<CartInfo> sumStream = carInfoStream
            .keyBy("sensorId")
            .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
            .sum("count");

// 4. 数据终端-sink
    sumStream.printToErr();
// 5. 应用执行
env.execute(StreamSlidingTimeWindow_3.class.getSimpleName());
} }